package com.bw.yk10.func;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Properties;

public class HbaseSinkFunction extends RichSinkFunction<String> {

    private static Connection con;

    public static void initConnection() {
        try {

//            Properties props = new Properties();
//            props.setProperty("phoenix.schema.isNamespaceMappingEnabled", "true");
            Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
            con=DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
//            con=DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181",props);
//            con.setAutoCommit(true);
        }catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        if(con==null)
            initConnection();
    }


    @Override
    public void invoke(String value, Context context) throws Exception {
        JSONObject jsonObject = (JSONObject) JSONObject.parse(value);


        String table=jsonObject.getString("table");
        String id=jsonObject.getObject("data",JSONObject.class).getString("id");;
        String line=jsonObject.getObject("data",JSONObject.class).toJSONString();

        String sql = "UPSERT INTO dim_"+table+" (id,line) VALUES (?,?)";

        PreparedStatement ps =null;
        
        if(con!=null)
        {
            try {
                ps=con.prepareStatement(sql);
                ps.setString(1,id);
                ps.setString(2,line);

                ps.execute();
                con.commit();
            }catch (Exception e)
            {
                e.printStackTrace();
            }finally {
                if(ps!=null)
                    ps.close();
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(con!=null)
        {
            try {
                con.close();
            }catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }
}
